#include "connection.h"
#include "log.h"
#include "message_parser.h"

#include "singleton.h"
#include "message_builder.h"

namespace edm{
namespace network{

connection_t::connection_t(boost::asio::io_service& io_service):
m_io_service(io_service),m_strand(io_service),m_resolver(io_service),m_socket(io_service),
m_host(), m_port(),m_mode(E_CON_PASSIVE),
m_connected(false),m_writting_flag(false), m_callback_handler(0),m_callback_error(0),m_callback_connect(0)
{
    logtrace((MYNET, "connection_t::connection_t in passive mode"));
}

connection_t::connection_t(boost::asio::io_service& io_service,
                      const std::string& host, const std::string& port):
        m_io_service(io_service),m_strand(io_service),m_resolver(io_service),m_socket(io_service),
        m_host(host), m_port(port),m_mode(E_CON_ACTIVE),
        m_connected(false),m_callback_handler(0),m_callback_error(0)
{
    logtrace((MYNET, "connection_t::connection_t in active mode"));
}

connection_t::~connection_t()
{
    logtrace((MYNET, "connection_t::~connection_t"));
    if(m_connected) stop();
    if(!m_write_messages.empty())
    {
        logwarn((MYNET, "connection_t::~connection_t left <%d> messages not delivery ...", m_write_messages.size()));
    }
    logtrace((MYNET, "connection_t::~connection_t"));
}

boost::asio::ip::tcp::socket& connection_t::socket()
{
    return m_socket;
}

bool connection_t::connected()
{
    return m_connected;
}

//! get ip
std::string connection_t::remote_address()
{
    return m_remote_addr;
}

std::string connection_t::local_address()
{
    return m_host;
}
            
//! default async read from connection, active mode will just connect first
void connection_t::start(callback_handler_t cb, callback_error_t cb_err, callback_connect_t cb_ok)
{
    logtrace((MYNET, "connection_t::start begin ..."));
    
    m_callback_handler = cb;
    m_callback_error = cb_err;
    if(cb_ok)
    {
       m_callback_connect = cb_ok; 
    }
    //! active connect to host
    if(m_mode == E_CON_ACTIVE && !m_connected)
    {
        boost::asio::ip::tcp::resolver::query query(m_host, m_port);
        m_resolver.async_resolve(query,
                                 boost::bind(&connection_t::handle_resolve, shared_from_this(),
                                 boost::asio::placeholders::error,
                                 boost::asio::placeholders::iterator));
        
        return;
    }
    
    if(m_socket.is_open()) 
    {
        m_connected = true;
        m_remote_addr = m_socket.local_endpoint().address().to_string();
                
        if(m_callback_connect && m_mode == E_CON_PASSIVE)
           m_callback_connect(shared_from_this());
        
        trigger_read();           
        trigger_write();
    }
    
    logtrace((MYNET, "connection_t::start end ok."));           
}

//! close connection
void connection_t::stop()
{
    logtrace((MYNET, "connection_t::stop begin ..."));   
    if(m_connected) m_socket.close();
    m_connected = false;
    logtrace((MYNET, "connection_t::stop end ok."));   
}

//! get cb hander
callback_handler_t connection_t::callback_handler()
{
    return m_callback_handler;
}
//! get cb error
callback_error_t connection_t::callback_error()
{
    return m_callback_error;
}

//! get cb connect
callback_connect_t connection_t::callback_connect()
{
    return m_callback_connect;
}
            
void connection_t::trigger_read()
{
    logtrace((MYNET, "connection_t::trigger_read begin ..."));
    boost::asio::async_read(m_socket,
                            boost::asio::buffer(m_read_header_buffer, sizeof(message_header_t)),
                            m_strand.wrap(boost::bind(
                                                    &connection_t::handle_read_header, shared_from_this(),
                                                    boost::asio::placeholders::error,
                                                    boost::asio::placeholders::bytes_transferred)));
    logtrace((MYNET, "connection_t::trigger_read end ok."));
}

void connection_t::trigger_write()
{
    logtrace((MYNET, "connection_t::trigger_write begin ..."));
    if(!m_write_messages.empty() && m_connected)
    {
        char hdr[MESSAGE_HDR_MAX_LEN] = {0};
        message_builder_t::build_header(m_write_messages.front().header, hdr, MESSAGE_HDR_MAX_LEN);
        m_writting_raw.assign(hdr, sizeof(message_header_t));
        m_writting_raw.append(m_write_messages.front().body);
        m_writting_flag = true;
        boost::asio::async_write(m_socket,
                                 boost::asio::buffer(m_writting_raw.data(), m_writting_raw.length()),
                                 m_strand.wrap(boost::bind(&connection_t::handle_write, shared_from_this(),
                                 boost::asio::placeholders::error)));
    }   
    logtrace((MYNET, "connection_t::trigger_write end ok."));
}

//! resolver
void connection_t::handle_resolve(const boost::system::error_code& error, 
           boost::asio::ip::tcp::resolver::iterator endpoint_iterator)
{
    logtrace((MYNET, "connection_t::handle_resolve begin ..."));   
    if(error)
    {
        logerror((MYNET, "connection_t::handle_resolve failed:<%s>", error.message().c_str()));
        handle_error(error);
        return;    
    }
    boost::asio::ip::tcp::endpoint endpoint = *endpoint_iterator;
    m_socket.async_connect(endpoint,
          boost::bind(&connection_t::handle_connect, shared_from_this(),
            boost::asio::placeholders::error, ++endpoint_iterator));
    logtrace((MYNET, "connection_t::handle_resolve end ok."));   
}

//! connect
void connection_t::handle_connect(const boost::system::error_code& error,
           boost::asio::ip::tcp::resolver::iterator endpoint_iterator)
{
    logtrace((MYNET, "connection_t::handle_connect begin ..."));   
    if(error)
    {
        logerror((MYNET, "connection_t::handle_connect failed:<%s>", error.message().c_str()));
        handle_error(error);
        return;    
    }
    m_connected = true;
    m_remote_addr = m_socket.local_endpoint().address().to_string();
    if(m_callback_connect && m_mode == E_CON_ACTIVE)
        m_callback_connect(shared_from_this());
    
    trigger_read();
    trigger_write();
    logtrace((MYNET, "connection_t::handle_connect end ok.")); 
}

void connection_t::async_write(COMMANDER command, edm_uint64_t reserved, const char*  body)
{
    logtrace((MYNET, "connection_t::async_write"));
    std::string sbody(body);
    m_io_service.post(boost::bind(&connection_t::sync_write_i, shared_from_this(),
                                  command, reserved, sbody));
}

//! write message
void connection_t::async_write(edm::common::COMMANDER command, edm_uint64_t reserved, std::string& body)
{
    logtrace((MYNET, "connection_t::async_write"));
    m_io_service.post(boost::bind(&connection_t::sync_write_i, shared_from_this(),
                                  command, reserved, body));
}
void connection_t::sync_write_i(COMMANDER command, edm_uint64_t reserved, std::string& body)
{
    logtrace((MYNET, "connection_t::sync_write_i connected <%s> begin ...",
                m_connected ? "true": "false"));
    loginfo((MYNET, "MYNET: message response <%d:%s>", command, body.c_str()));
    bool empty = m_write_messages.empty();
    message_t message;
    message.header.command = (message_header_command_t)command;
    message.header.reserve = reserved;
    //message.header.len = body.length();
    message.header.len = body.length() + sizeof(message_header_t);
    message.body = body;
    m_write_messages.push_back(message);
    if(empty) trigger_write();
    logtrace((MYNET, "connection_t::sync_write_i end ok."));
}
//! reader header cb
void connection_t::handle_read_header(const boost::system::error_code& error, std::size_t bytes_transferred)
{
    logtrace((MYNET, "connection_t::handle_read_header callback begin ..."));
    
    if(error)
    {
        if(m_mode == E_CON_PASSIVE && error == boost::asio::error::eof)
        {
            logtrace((MYNET, "connection_t::handle_read_header <peer:%s> end ok.", error.message().c_str()));
            handle_error(error);
            return;
        }
        logerror((MYNET, "connection_t::handle_read_header failed:<%s>", error.message().c_str()));
        handle_error(error);
        return;
    }
    
    if(bytes_transferred != sizeof(message_header_t))
    {
        logerror((MYNET, "connection_t::handle_read_header len not enough"));
        handle_error(error);
        return;        
    }
    
    //! parse header failed
    if (message_parser_t::parse_header(m_read_header_buffer, 
                                       sizeof(message_header_t), 
                                       m_read_message.header))
    {
        logerror((MYNET, "connection_t::handle_read_header => parse header failed"));
        handle_error(error);
        return;
    }
    logdebug((MYNET, "connection_t::handle_read_header <%d:%d>", m_read_message.header.command,
                                                                 m_read_message.header.len));
    
    //! only header
    if(m_read_message.header.len == 0)
    {
        if(m_callback_handler)
        {
            logdebug((MYNET, "connection_t::handle_read_body callback handler"));
            loginfo((MYNET, "MYNET: message request:<%d:%s>", m_read_message.header.command, m_read_message.body.c_str()));
            m_callback_handler(shared_from_this(), m_read_message);
        }
        else
            logdebug((MYNET, "connection_t::handle_read_body callback handler is zero"));
        memset(&m_read_message.header, 0, sizeof(message_header_t));
        m_read_message.body.clear();
        trigger_read();
        logtrace((MYNET, "connection_t::handle_read_header callback end ok."));
        return;
    }
    
    m_read_left_len = (int)m_read_message.header.len;
    size_t will_read_len = m_read_left_len;
    if(will_read_len > READ_BUFFER_LEN) will_read_len = READ_BUFFER_LEN;
    boost::asio::async_read(m_socket,
                            boost::asio::buffer(m_read_body_buffer, will_read_len),
                            m_strand.wrap(boost::bind(
                            &connection_t::handle_read_body, shared_from_this(),
                            boost::asio::placeholders::error,
                            boost::asio::placeholders::bytes_transferred)));
    
    /*m_socket.async_read_some(boost::asio::buffer(m_read_body_buffer),
                             boost::bind(
                             &connection_t::handle_read_body, 
                             shared_from_this(),
                             boost::asio::placeholders::error,
                             boost::asio::placeholders::bytes_transferred));
     */
    logtrace((MYNET, "connection_t::handle_read_header callback end ok."));
}

//! reader body cb
void connection_t::handle_read_body(const boost::system::error_code& error, std::size_t bytes_transferred)
{
    logtrace((MYNET, "connection_t::handle_read_body callback begin ..."));
    if(error)
    {
        if(m_mode == E_CON_PASSIVE && error == boost::asio::error::eof)
        {
            logtrace((MYNET, "connection_t::handle_read_header <peer:%s> end ok.", error.message().c_str()));
            handle_error(error);
            return;
        }        
        logerror((MYNET, "connection_t::handle_read_body failed:<%s>", error.message().c_str()));
        handle_error(error);
        return;
    }    
    
    
    m_read_left_len = m_read_left_len - bytes_transferred;
    m_read_message.body.append(m_read_body_buffer, bytes_transferred);
    logdebug((MYNET, "m_read_left_len : %d", m_read_left_len));
    if(m_read_left_len > 0)
    {
        size_t will_read_len = m_read_left_len;
        if(will_read_len > READ_BUFFER_LEN) will_read_len = READ_BUFFER_LEN;
        boost::asio::async_read(m_socket,
                            boost::asio::buffer(m_read_body_buffer, will_read_len),
                            m_strand.wrap(boost::bind(
                            &connection_t::handle_read_body, shared_from_this(),
                            boost::asio::placeholders::error,
                            boost::asio::placeholders::bytes_transferred)));
        logtrace((MYNET, "connection_t::handle_read_body callback message not finished, left <%d> continue ...",
                 m_read_left_len));
        return;
    }
    
    //! message completed, trigger the callback handler
    if(m_callback_handler)
    {
        logdebug((MYNET, "connection_t::handle_read_body callback handler"));
        loginfo((MYNET, "MYNET: message request:<%d:%s>", m_read_message.header.command, m_read_message.body.c_str()));
        m_callback_handler(shared_from_this(), m_read_message);
    }
    else
        logdebug((MYNET, "connection_t::handle_read_body callback handler is zero"));
    memset(&m_read_message.header, 0, sizeof(message_header_t));
    m_read_message.body.clear();
    
    //! message completed, continue read header
    trigger_read();  
    logtrace((MYNET, "connection_t::handle_read_body callback end ok."));
}

//! write message cb
void connection_t::handle_write(const boost::system::error_code& error)
{
    logtrace((MYNET, "connection_t::handle_write callback begin ..."));
    if(error)
    {
        if(m_mode == E_CON_PASSIVE && error == boost::asio::error::eof)
        {
            logtrace((MYNET, "connection_t::handle_read_header <peer:%s> end ok.", error.message().c_str()));
            handle_error(error);
            return;
        }        
        logerror((MYNET, "connection_t::handle_write failed:<%s>", error.message().c_str()));
        handle_error(error);
        return;
    }

    m_write_messages.pop_front();
    trigger_write();
    logtrace((MYNET, "connection_t::handle_write callback end ok."));
}
            
//! handle error occur
void connection_t::handle_error(const boost::system::error_code& error)
{
    logtrace((MYNET, "connection_t::handle_error begin ..."));
    
    logerror((MYNET, "connection_t::handle_error occur:%s", error.message().c_str()));
    stop();
    if(m_callback_error) m_callback_error(shared_from_this());
    
    logtrace((MYNET, "connection_t::handle_error end ok."));
}


}}

